<?php
/**
 * @copyright  Copyright  LexisNexis, a division of Reed Elsevier Inc. All rights reserved.
 */
namespace sdk;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

include_once(__DIR__ . '/config.php');

class QueuePublisher {
    private static $obj;
    private $exchangeName;
    private $queueName;
    /**
     * @var AMQPStreamConnection
     */
    private $connection;
    /**
     * @var AMQPChannel
     */
    private $channel;

    public static function instance() {
        if (!self::$obj) {
            self::$obj = new self();
        }

        return self::$obj;
    }

    private function __construct() {
        return;
    }

    public function connect() {
        $this->exchangeName = MQ_EXCHANGE_NAME;
        $this->queueName = MQ_QUEUE_NAME;

        $this->connection = new AMQPStreamConnection(
            MQ_HOST, MQ_PORT, MQ_USER, MQ_PASS, MQ_VHOST
        );
        $this->channel = $this->connection->channel();
        //passive:是否被动，主动会自动建队列 durable:持久化 exclusive:是否私有 auto_delete:取消订阅自动删除队列
        $this->channel->queue_declare($this->queueName, false, true, false, false);
        $this->channel->exchange_declare($this->exchangeName, 'direct', false, true, false);
        $this->channel->queue_bind($this->queueName, $this->exchangeName);
    }

    public function push($messageArray = array()) {
        $this->connect();

        if (empty($messageArray['id'])) {
            $messageArray['id'] = md5(uniqid("", true));
        }
        $messageBody = json_encode($messageArray);
        $message = new AMQPMessage($messageBody, array(
            'content_type' => 'text/plain',
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
        ));
        $this->channel->basic_publish($message, $this->exchangeName);

        $this->close();
        return true;
    }

    public function close() {
        $this->channel->close();
        $this->connection->close();
    }
}